package org.micro.frame.es;

import lombok.extern.slf4j.Slf4j;
import org.apache.http.Header;
import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.message.BasicHeader;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
import org.springframework.boot.context.properties.EnableConfigurationProperties;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.util.StringUtils;

import java.util.Map;

/**
 * The Elasticsearch Auto Configuration
 *
 * @author lry
 * @since 1.0.0
 */
@Slf4j
@Configuration
@EnableConfigurationProperties(ElasticsearchProperties.class)
public class ElasticsearchAutoConfiguration {

    private final ElasticsearchProperties elasticsearchProperties;

    public ElasticsearchAutoConfiguration(ElasticsearchProperties elasticsearchProperties) {
        this.elasticsearchProperties = elasticsearchProperties;
    }

    @Bean
    @ConditionalOnMissingBean
    public RestHighLevelClient restHighLevelClient() {
        String[] addressNodes = elasticsearchProperties.getAddressNodes();
        HttpHost[] httpHosts = new HttpHost[addressNodes.length];
        for (int i = 0; i < addressNodes.length; i++) {
            httpHosts[i] = HttpHost.create(addressNodes[i]);
        }

        RestClientBuilder restClientBuilder = RestClient.builder(httpHosts);

        // setter username and password
        if (!StringUtils.isEmpty(elasticsearchProperties.getUserName())
                && !StringUtils.isEmpty(elasticsearchProperties.getPassword())) {
            final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
            credentialsProvider.setCredentials(AuthScope.ANY, new UsernamePasswordCredentials(
                    elasticsearchProperties.getUserName(), elasticsearchProperties.getPassword()));
            restClientBuilder.setHttpClientConfigCallback((httpAsyncClientBuilder) ->
                    httpAsyncClientBuilder.setDefaultCredentialsProvider(credentialsProvider));
        }

        // setter max retry timeout millis
        if (elasticsearchProperties.getMaxRetryTimeout() != null) {
            restClientBuilder.setMaxRetryTimeoutMillis(elasticsearchProperties.getMaxRetryTimeout());
        }

        // setter path prefix
        if (!StringUtils.isEmpty(elasticsearchProperties.getPathPrefix())) {
            restClientBuilder.setPathPrefix(elasticsearchProperties.getPathPrefix());
        }

        // setter header parameters
        Map<String, String> headers = elasticsearchProperties.getHeaders();
        if (headers != null) {
            Header[] tempHeaders = new Header[headers.size()];
            int k = 0;
            for (Map.Entry<String, String> entry : headers.entrySet()) {
                tempHeaders[k++] = new BasicHeader(entry.getKey(), entry.getValue());
            }
            if (tempHeaders.length > 0) {
                restClientBuilder.setDefaultHeaders(tempHeaders);
            }
        }

        log.info("The RestClientBuilder is: {}", restClientBuilder.toString());

        return new RestHighLevelClient(restClientBuilder);
    }

}
